package connection

import (
	"broker/interface/imodel"
	"broker/interface/iserver"
	"broker/struct/model"
	"broker/utils"
	"broker/utils/logger"
	"fmt"
	"sort"
	"sync"
)

/**
 * @Description
 * @Author huan.wang
 * @Date 2021/4/26 6:53 下午
 **/

type TopicManager struct {
	topics    map[string][]imodel.ISubscriber
	topicLock sync.RWMutex
}

var topicManger = &TopicManager{}
var topicManagerOnce sync.Once

func NewTopicManager() *TopicManager {
	topicManagerOnce.Do(func() {
		topicManger = &TopicManager{
			topics: make(map[string][]imodel.ISubscriber),
		}
	})
	return topicManger
}

func (tm *TopicManager) Publish(message model.PublishMessage, connection iserver.IConnection) error {
	tm.topicLock.RLock()
	go connection.GetTcpServer().CallOnPublish(connection,message)
	filters := make([]string, 0, len(tm.topics))
	for key := range tm.topics {
		if utils.CheckTopic(message.GetTopicString(), key) {
			filters = append(filters, key)
		}
	}
	sendSubscribers := make([]imodel.ISubscriber, 0)
	for _, filter := range filters {
		sendSubscribers = append(sendSubscribers, tm.topics[filter]...)
	}
	tm.topicLock.RUnlock()

	for _, subscriber := range sendSubscribers {
		conn, err := connection.GetTcpServer().GetConnManager().GetConnection(subscriber.GetConnID())
		if err != nil {
			logger.Error.Println("[Publish] publish topic ", message.GetTopicString(), " on ", subscriber.GetConnID(), " get connection error", err)
			continue
		}
		err = conn.SendMsg(message)
		if err != nil {
			logger.Error.Println("[Publish] publish topic ", message.GetTopicString(), " on ", subscriber.GetConnID(), " send message error", err)
			continue
		}
		logger.Info.Println(fmt.Sprintf("推送给%d",subscriber.GetConnID()))

	}
	return nil
}

func (tm *TopicManager) Subscribe(subscriber imodel.ISubscriber) error {
	tm.topicLock.Lock()
	defer tm.topicLock.Unlock()
	for _, topicFilter := range subscriber.GetTopicFilters() {
		if _, ok := tm.topics[topicFilter.GetFilter()]; ok {
			//已存在该主题过滤器的订阅队列
			index := sort.Search(len(tm.topics[topicFilter.GetFilter()]), func(i int) bool {
				return tm.topics[topicFilter.GetFilter()][i].GetConnID() >= subscriber.GetConnID()
			})
			if index == len(tm.topics[topicFilter.GetFilter()]) {
				tm.topics[topicFilter.GetFilter()] = append(tm.topics[topicFilter.GetFilter()], subscriber)
				sort.Slice(tm.topics[topicFilter.GetFilter()], func(i, j int) bool {
					return tm.topics[topicFilter.GetFilter()][i].GetConnID() < tm.topics[topicFilter.GetFilter()][j].GetConnID()
				})
			}
		} else {
			//不存在该主题订阅队列->新建订阅者队列并加入
			tm.topics[topicFilter.GetFilter()] = make([]imodel.ISubscriber, 0)
			tm.topics[topicFilter.GetFilter()] = append(tm.topics[topicFilter.GetFilter()], subscriber)
		}
	}
	logger.Info.Println(fmt.Sprintf("%d订阅%s成功",subscriber.GetConnID(),subscriber.GetTopicFilters()[0].GetFilter()))
	return nil
}

func (tm *TopicManager) RemoveSubscribe(rmsubscriber imodel.ISubscriber) error {
	tm.topicLock.Lock()
	defer tm.topicLock.Unlock()
	for key := range tm.topics {

		index := sort.Search(len(tm.topics[key]), func(i int) bool {
			return tm.topics[key][i].GetConnID() >= rmsubscriber.GetConnID()
		})
		if index >= len(tm.topics) {
			continue
		}
		logger.Info.Println(fmt.Sprintf("需要删除的连接是%d",rmsubscriber.GetConnID()))
		logger.Info.Println(fmt.Sprintf("找到的下标是%d",index))
		logger.Info.Println(fmt.Sprintf("数组长度是%d",len(tm.topics[key])))
		logger.Info.Println(tm.topics)
		//if index == len(tm.topics) - 1{
		//	if index == 0 && len(tm.topics)  == 1 {
		//		delete(tm.topics, key)
		//		return nil
		//	}else{
		//		tm.topics[key] = tm.topics[key][0:index]
		//	}
		//
		//}else{
		//	tm.topics[key] = append(tm.topics[key][:index], tm.topics[key][index+1:]...)
		//}
		//
		//if len(tm.topics[key]) == 0 {
		//	delete(tm.topics, key)
		//}

	}
	return nil
}

func (tm *TopicManager) UnSubscribe(unsubscriber imodel.ISubscriber) error {
	tm.topicLock.Lock()
	defer tm.topicLock.Unlock()
	for _, unsub := range unsubscriber.GetTopicFilters() {
		for key := range tm.topics {
			if unsub.GetFilter() == key {
				index := sort.Search(len(tm.topics[key]), func(i int) bool {
					return tm.topics[key][i].GetConnID() >= unsubscriber.GetConnID()
				})
				if index >= len(tm.topics) {
					continue
				}
				tm.topics[key] = append(tm.topics[key][:index], tm.topics[key][index+1:]...)
				if len(tm.topics[key]) == 0 {
					delete(tm.topics, key)
				}
			}
		}
	}
	logger.Info.Println(fmt.Sprintf("%d取消订阅成功",unsubscriber.GetConnID()))
	return nil
}
